package elephant.rpc.server.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import elephant.rpc.server.RPCServer;
import elephant.rpc.server.message.RPCMessage;
import elephant.rpc.server.message.RPCRequest;
import elephant.rpc.server.service.RPCServiceManager;
import elephant.rpc.server.session.RPCSession;
import elephant.rpc.server.session.RPCSessionManager;
import elephant.rpc.threadpool.ThreadPoolManager;

/**
 * 
 */
public abstract class NetworkService{
	//
	private static Logger logger=LoggerFactory.getLogger(NetworkService.class);
	//
	protected RPCServer rpcServer;
	protected RPCSessionManager sessionManager;
	protected RPCServiceManager serviceManager;
	protected ThreadPoolManager threadPoolManager;
	protected PerformMonitor performMonitor;
	
	public NetworkService(RPCServer rpcServer){
		this.rpcServer=rpcServer;
		this.sessionManager=rpcServer.getSessionManager();
		this.serviceManager=rpcServer.getServiceManager();
		this.threadPoolManager=rpcServer.getThreadPoolManager();
		this.performMonitor=rpcServer.getPerformMonitor();
	}
	
	/**
	 * 
	 * @throws Exception
	 */
	public abstract void init() throws Exception;
	
	/**
	 * 
	 * @throws Exception
	 */
	public abstract void start() throws Exception;
	
	/**
	 * 
	 * @throws Exception
	 */
	public abstract void stop() throws Exception;
	
	/**
	 * 
	 */
	public void onSessionCreate(RPCSession session){
		if(logger.isDebugEnabled()){
			logger.info("rpc session create:{}",session);
		}
	}
	
	/**
	 * 
	 */
	public void onSessionDisconnected(RPCSession session){
		if(logger.isDebugEnabled()){
			logger.info("rpc session disconnected:{}",session);
		}
		sessionManager.sessionDisconnected(session);
	}
	
	/**
	 * 
	 * @param session
	 * @param message
	 */
	public void onMessageReceived(RPCSession session,RPCMessage message) {
		session.lastAccessTime=System.currentTimeMillis();
		switch (message.type) {
		case RPCMessage.TYPE_REQ:
			onReqMessageReceived(session,message);
			break;
		case RPCMessage.TYPE_AUTH:
			onAuthMessageReceived(session,message);
			break;
		default:
			logger.error("bad message type:"+message.type);
			break;
		}
	}
	//
	private void onAuthMessageReceived(RPCSession session,RPCMessage message) {
		synchronized (session) {
			String clientUuid=(String)message.payloads[0];
			String sign=(String)message.payloads[1];
			if(!RPCMessage.AUTH_SIGN.equals(sign)){
				logger.error("auth failed.sign error {}",sign);
				session.disconnect();
				return;
			}
			if(message.payloads.length>2){
				for(int i=2;i<message.payloads.length;i++){
					session.topics.add(message.payloads[i].toString());
				}
			}
			session.clientUuid=clientUuid;
			sessionManager.sessionCreated(session);	
			//
			if(logger.isDebugEnabled()){
				logger.debug("onAuth Success {}",session);
			}
		}
	}
	//
	private void onReqMessageReceived(RPCSession session, RPCMessage message) {
		RPCRequest req=new RPCRequest();
		req.setMessage((RPCMessage) message);
		req.setSession(session);
		RPCThreadWorker worker=new RPCThreadWorker(performMonitor,serviceManager,req);
		threadPoolManager.execute(worker);
	}
}
